package com.zhou.common.mybatis.template.oss.mq.consumer;

import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.OrderConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.zhou.common.base.mq.mq.MqConfig;
import com.zhou.common.mybatis.template.oss.mq.consumer.listener.OssKeysListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @author fenghua
 * @version 1.0
 * @date 2021/6/17 9:43 上午
 */

@ConditionalOnProperty(prefix = "rocketmq", name = "enable", havingValue = "false", matchIfMissing = true)
@Configuration
public class CaseInfoConsumerClient {

    @Autowired
    private MqConfig mqConfig;

    @Autowired
    private MqConsumerProperties mqConsumerProperties;

    @Value("${flow.task.topic}")
    private String flowTaskTopic;

    @Autowired
    private OssKeysListener messageListener;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public OrderConsumerBean buildConsumer() {
        OrderConsumerBean consumer = new OrderConsumerBean();
        Properties properties = mqConfig.getMqProperties();
        properties.setProperty(PropertyKeyConst.GROUP_ID, mqConsumerProperties.getGroupId());
        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "5");
        properties.setProperty(PropertyKeyConst.MaxReconsumeTimes, "2");
        consumer.setProperties(properties);

        // 订阅关系
        Map<Subscription, MessageOrderListener> subscriptionTable = new HashMap<>();

        Subscription acceptSubscription = new Subscription();
        acceptSubscription.setTopic(mqConsumerProperties.getTopic());
        subscriptionTable.put(acceptSubscription, messageListener);

        consumer.setSubscriptionTable(subscriptionTable);

        return consumer;
    }


}
